package com.bbzn.device.client.service.emq;

import com.alibaba.fastjson.JSONObject;
import com.bbzn.device.client.dataobject.Strategy;
import com.bbzn.device.client.service.CwtingMessageStrategy;
import com.bbzn.device.client.service.PublishService;
import com.bbzn.device.client.service.StrategyService;
import com.bbzn.device.client.utils.SpringUtils;
import com.ctiot.aep.mqmsgpush.sdk.IMsgConsumer;
import com.ctiot.aep.mqmsgpush.sdk.IMsgListener;
import com.ctiot.aep.mqmsgpush.sdk.MqMsgConsumer;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;

/**
 * @Author wulongbo
 * @Date 2021/8/30 10:07
 * @Version 1.0
 */


@Service
@Slf4j
public class CtwingMqSubService {

    @Autowired
    private StrategyService strategyService;

    @Autowired
    private PublishService publishService;

    private String server = "msgpush.ctwing.cn:16651"; //消息服务地址
    private String tenantId = "10525760";//租户ID
    private String token = "eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiIxMDUyNTc2MCJ9.nIa9Aa-xBqclk7NYECZ5ufZupsVMxgO9pDEa7_jNNPs";//身份认证token串
    private String certFilePath = ""; //直接填空字符串，CA证书，JDK已经内置相关根证书，无需指定

    @PostConstruct
    public void init() {
        //创建消息接收Listener
        IMsgListener msgListener = new IMsgListener() {
            @SneakyThrows
            @Override
            public void onMessage(String msg) {
                //接收消息
                log.info("msg arrive:" + msg);
                //消息处理...
                //为了提高效率，对消息进行异步处理（使用其它线程、发送到Kafka等）
                JSONObject jsonObject = JSONObject.parseObject(msg);
                String messageType = jsonObject.getString("messageType");
                /* * 电信订阅推送消息格式
                 *   消息格式描述:
                 *   1.设备数据变化  dataReport
                 *   2.设备命令响应  commandResponse
                 *   3.设备事件上报  eventReport
                 *   4.设备上下线   deviceOnlineOfflineReport
                 *   5.TUP合并数据变化  dataReportTupUnion
                 * */
                if (StringUtils.isNotEmpty(messageType)) {
                    // 2.根据code查询具体业务实现
                    Strategy strategy = new Strategy();
                    strategy.setStrategyEntityId(messageType);
                    strategy = strategyService.getStrategy(strategy);
                    if (null == strategy) {
                        return;
                    }
                    String strategyBeanId = strategy.getStrategyBeanId();
                    if (null == strategyBeanId) {
                        return;
                    }
                    CwtingMessageStrategy strategyService = SpringUtils.getBean(strategyBeanId, CwtingMessageStrategy.class);
                    strategyService.pushMessage(jsonObject);
                }
            }
        };

        //创建消息接收类
        IMsgConsumer consumer = new MqMsgConsumer();
        try {

            //初始化
            /*        * @param server  消息服务server地址
             * @param tenantId 租户Id
             * @param token    用户认证token
             * @param certFilePath 证书文件路径
             * @param topicNames   主题名列表，如果该列表为空或null，则自动消费该租户所有主题消息
             * @param msgListener 消息接收者
             * @return 是否初始化成功
             * */
            consumer.init(server, tenantId, token, certFilePath, null, msgListener);

            //开始接收消息
            consumer.start();
            //程序退出时，停止接收、销毁
            //consumer.stop();
            //consumer.destroy();
        } catch (
                Exception e) {
            e.printStackTrace();
        }
        System.out.println("exit");
    }
}
